package com.itao.rocketmq5.delay;

import com.itao.rocketmq5.util.RocketMqFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;

/**
 * @Author sjt
 * @CreateTime 2023/10/10 14:12
 */
@Slf4j
public class DelayConsumer {

    private final static String DELAY_TOPIC = "delay_topic";
    private final static String DELAY_CONSUMER_GROUP = "delay_consumer_group";

    public static void main(String[] args) {
        var expression = new FilterExpression("*", FilterExpressionType.TAG);

        try (
                var consumer = RocketMqFactory.provider().newPushConsumerBuilder()
                        .setClientConfiguration(RocketMqFactory.configuration())
                        .setConsumerGroup(DELAY_CONSUMER_GROUP)
                        .setSubscriptionExpressions(Collections.singletonMap(DELAY_TOPIC, expression))
                        .setMessageListener(messageView -> {
                            log.info("message: {}", StandardCharsets.UTF_8.decode(messageView.getBody()));
                            return ConsumeResult.SUCCESS;
                        })
                        .build()
        ) {
            log.info("消费者启动：{}", consumer);
            Thread.sleep(Long.MAX_VALUE);
        } catch (ClientException | IOException | InterruptedException e) {
            log.error("异常：", e);
        }
    }
}
